package com.foreveross.taskservice.engine.worker;

import java.util.Collection;
import java.util.List;
import java.util.Set;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import com.foreveross.taskservice.common.CacheDataSpace;
import com.foreveross.taskservice.common.TaskStatusEnum;
import com.foreveross.taskservice.common.model.AbstractTaskModel;
import com.foreveross.taskservice.core.domain.channel.GradChannelEntity;
import com.foreveross.taskservice.infra.cfg.PropertyUtils;
import com.foreveross.taskservice.infra.cfg.taskservice.TaskServicePropertiesLoader;
import com.google.common.collect.Lists;

/**
 * 将任务推送入MQ的操作
 * 
 * @author Administrator
 * 
 */
public class TaskPushToMQWorker extends AbstractTaskStatefulJobWorker {
	@Override
	public void execute(JobExecutionContext context)
			throws JobExecutionException {
		long stime = System.currentTimeMillis();
		log.info("任务发布TaskPushToMQWorker开始....");
		List<GradChannelEntity> channels = null;
		try {
			int queueMaxSize = PropertyUtils.getIntProperty(
					"taskservice.mq.queue.maxsize",
					TaskServicePropertiesLoader.FILE_NAME, 50);
			channels = getGradChannelApplication().getEnableChannel();
			CacheDataSpace space=CacheDataSpace.AirFreightSpace();
			for(GradChannelEntity channel:channels){
				// 获取MQ上队列的长度，如果不足maxSize,则补充
				int free = queueMaxSize - (int) getMqInfoApplication().getTaskQueueSize(space, channel.getHbaseKey());
				//取用之前先查询避免直接加锁浪费资源
				if (free > 0 && getTaskCacheQuery().queryTaskSize(space, channel.getHbaseKey(), TaskStatusEnum.TASK_WATI)>0) {
//					Set<AbstractTaskModel> tasks = getTaskCacheStorage().getTasks(space,channel.getHbaseKey(), free);
					//新redis之后的模型查找(airfeightmodel)
					List<AbstractTaskModel> tasks=getTaskCacheStorage().getTasks(channel.getHbaseKey(), free, TaskStatusEnum.TASK_WATI.getStatus()+"");
					if (CollectionUtils.isNotEmpty(tasks)) {
						clearAction(tasks);
						getTaskToMQApplication().sendTasks(tasks);
						log.info(String.format("渠道[%s]推送任务[%s]", channel.getName(),tasks.size()));
					}
				}
			}
		} catch (Exception e) {
			log.error(ExceptionUtils.getFullStackTrace(e));
		}finally{
			channels = null;
			log.info("任务发布TaskPushToMQWorker 结束....耗时ms:"+(System.currentTimeMillis()-stime));
		}
	}
	/**
	 * 推送到MQ之前将任务里面的action进行清空，原因
	 * 1 减少网络流量带宽
	 * 2 节点不会关心action,仅仅会产生新的记录
	 * 	 我们只需要回滚回来的时候和内存里面的进行任务进行合并就可以了，通过接口定义的方法实现
	 * @param models
	 */
	private void clearAction(Collection<AbstractTaskModel> models){
		for(AbstractTaskModel m:models){
			m.clearActions();
		}
	}

}
